/* Pipe.java
 *
 * Author:	Malcolm Staudacher
 * Date:	Oct 1, 2008
 *
 */

package itb.fhv.sa.pipesfilters;

import java.util.LinkedList;
import java.util.NoSuchElementException;

public abstract class Pipe extends PLElement {

	protected LinkedList<PLOutput> m_buffer = null;
	protected int m_bufsize;
	protected boolean isSyncPipe;
	private boolean end = false;


	// needed for last pipe in a push pipeline
	public Pipe() {
		isSyncPipe = false;
		m_buffer = new LinkedList<PLOutput>();
	}


	public Pipe(Filter neighbour) {
		m_neighbour1 = neighbour;
		m_buffer = new LinkedList<PLOutput>();
		isSyncPipe = false;
	}


	/**
	 * Reads data from the pipe. If this is not a synchronized pipe, then this
	 * method will call the read method on the preceding pipeline element.
	 * 
	 * @return The next element of data from the buffer, or null if the buffer
	 *         is empty.
	 */
	@Override
	public PLOutput read() {
		PLOutput in = null;

		if (isSyncPipe == false) {

			while (m_buffer.size() < m_bufsize && end == false) {

				in = m_neighbour1.read();

				if (in != null) {

					// detect the end of the data
					if (in.getClass() != Character.class) {

						if (in.getContent().contains("\1")) {
							end = true;
						}

					}

					m_buffer.add(in);
				}

			}

		}

		synchronized (m_buffer) {

			if (m_buffer.size() > 0) {
				PLOutput plo = null;

				try {
					plo = m_buffer.pollFirst();

					if (plo != null && plo.getClass() != Character.class) {

						// detect the end of the data
						if (plo.getContent().contains("\1")) {
							end = true;
						}

					}

				} catch (NoSuchElementException e) {
					System.out.println("no first element");
				}

				return plo;
			} else {
				// wait for other thread to write to the buffer
				try {
					m_buffer.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}

				return null;
			}

		}

	}


	/**
	 * Writes to this pipe. If the pipe is not a synchronisation pipe, then this
	 * method calls the write method on the next pipeline element.
	 * 
	 * @param plo
	 *            The PLOutput data to write to this pipe.
	 * @return true, if the element was written to the buffer, false otherwise.
	 */
	@Override
	public boolean write(PLOutput plo) {

		synchronized (m_buffer) { // only one thread may access the buffer

			if (m_buffer.size() >= m_bufsize) {

				if (isSyncPipe == false) {
					// only write to next element, if this is not a sync pipe
					while (m_buffer.size() > 0) {
						m_neighbour1.write(m_buffer.pollFirst());
					}

				}

			} else {

				if (plo != null) {
					m_buffer.add(plo);
				}

				// if other threads are waiting for the buffer, they must be
				// notified
				m_buffer.notify();
				return true;
			}

			m_buffer.notify();
		}

		return false;
	}


	/**
	 * Used to propagate the end of the data.
	 */
	protected void flush() {

		if (isSyncPipe == false) {

			// empty the buffer
			while (m_buffer.size() > 0) {
				m_neighbour1.write(m_buffer.pollFirst());
			}

			m_neighbour1.flush();
		}

	}


	/**
	 * Sets this pipe as a synchronisation pipe.
	 * 
	 * @param isSyncPipe
	 *            sets this pipe as a sync pipe.
	 */
	public void setSyncPipe(boolean isSyncPipe) {
		this.isSyncPipe = isSyncPipe;
	}

}
